package org.databandtech.mockmq.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;

public class AsyncCommitConsumer {

    final KafkaConsumer<String, String> consumer;

    private volatile boolean isRunning = true;

    public AsyncCommitConsumer(KafkaConsumer<String, String> consumer,String topicName) {
    	this.consumer = consumer;
        consumer.subscribe(Arrays.asList(topicName));
    }

    public void printReceiveMsg() {
        try {
            while (isRunning) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                if (!consumerRecords.isEmpty()) {
                    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                        System.out.println("TopicName: " + consumerRecord.topic() + " Partition:" +
                                consumerRecord.partition() + " Offset:" + consumerRecord.offset() + "" +
                                " Msg:" + consumerRecord.value());
                        //进行逻辑处理
                    }
                    consumer.commitAsync();//异步提交
                }
            }
        }catch (Exception e){
            //处理异常
        }
        finally {
            consumer.commitAsync();
            close();
        }
    }

    public void close() {
        isRunning = false;
        if (consumer != null) {
            consumer.close();
        }
    }
}